package javaVersion.learn.stateProcess.broadCastStateDemo;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.*;
import java.util.HashMap;
import java.util.Map;

/**
 * @program: myFlink
 * @description: 构建Flink的mysql数据源
 * @author: WincoMa
 * @create: 2020-07-27 16:28
 **/
public class MysqlSource extends RichSourceFunction<Map<String,Integer>> {
    private String host = "localhost";
    private int port = 3306;
    private String db = "test";
    private String user = "root";
    private String passwd = "123456";
    private int secondInterval = 1;

    Connection connection = null;
    PreparedStatement ps = null;
    boolean isRunning = true;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        Class.forName("com.mysql.jdbc.Driver");
        Connection connection = DriverManager.getConnection("jdbc:mysql://" + host + ":" + port + "/" + db + "?useUnicode=true&characterEncoding=UTF-8", user, passwd);
        String sql = "select userID,userName,userAge from user_info";
        ps = connection.prepareStatement(sql);

    }

    @Override
    public void run(SourceContext<Map<String, Integer>> ctx) throws Exception {
        try {
            while (isRunning) {
                Map<String, Integer> output = new HashMap<String, Integer>();
                ResultSet rs = ps.executeQuery();
                while (rs.next()) {
                    String userID = rs.getString("userID");
                    String userName = rs.getString("userName");
                    int userAge = rs.getInt("userAge");
                    Map<String, Integer> map = new HashMap<>();
                    Integer info = map.put(userName, userAge);
                    output.put(userID, info);
                }
                ctx.collect(output);
                Thread.sleep(1000 * secondInterval);
            }
        } catch (Exception e) {
            System.out.println("从Mysql获取配置异常..."+ e);
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }

    @Override
    public void close() throws Exception {
        super.close();
        if (connection != null) {
            connection.close();
        }
        if (ps != null) {
            ps.close();
        }
    }
}
